package cn.itcast.spark.rdd

import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

class ActionOp {
  val conf = new SparkConf().setMaster("local[6]").setAppName("action_ope")
  val sc = new SparkContext(conf)

  /**
   * 需求，最终生成（“结果”, price）
   * 函数中传入的curr参数，并不是value，而是一整条数据
   * reduce整体上的结果，只有一个
   */
  @Test
  def reduce(): Unit = {
    val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
    val result = rdd.reduce((curr, agg) => ("总价", curr._2 + agg._2))
    println(result)
  }

  @Test
  def foreach(): Unit = {
    val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5))
    rdd.foreach(item => println(item))
  }

  /**
   * 每次调用action都会生成一个jon
   * countByKey的运算结果是Map(Key, Value -> Key的count)
   * 可为数据倾斜提供分析
   */
  @Test
  def count(): Unit = {
    val rdd = sc.parallelize(Seq(("a", 1), ("a", 2), ("c", 3), ("c", 4)))
    println(rdd.count())
    println(rdd.countByKey())
  }

  /**
   * first只会处理第一个分区的第一条数据，速度会比较快
   */
  @Test
  def take(): Unit = {
    val rdd = sc.parallelize(Seq(1, 2, 3, 4))
    rdd.take(3).foreach(item => println(item))
    println(rdd.first())
    rdd.takeSample(true, 3).foreach(item => println(item))
  }

}
